/*
 * Copyright (C), 2002-2021, 苏宁易购电子商务有限公司
 * FileName: KafkaConsumer.java
 * Author:   zhangc
 * Date:     2021/11/05 16:25
 * Description: //模块目的、功能描述
 * History: //修改记录
 * <author>      <time>      <version>    <desc>
 * 修改人姓名    修改时间    版本号       描述
 */
package com.zhangc.kafkatest.kafka;

import java.util.List;
import java.util.Optional;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * 功能描述:<br>
 *
 * @author zhangc
 * @see [相关类/方法]（可选）
 * @since [产品/模块版本] （可选）
 */
@Component
public class KafkaConsumer {

    private static final Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);

    //    //kafka的监听器，topic为"test_topic"，消费者组为"testGroup"
    //    @KafkaListener(topics = "test_topic", groupId = "testGroup")
    //    public void listenZhugeGroup(ConsumerRecord<String, String> record, Acknowledgment ack) {
    //        String value = record.value();
    //        logger.info("收到msg1:{}", value);
    //        logger.info("收到msg2:{}", record);
    //        //手动提交offset
    //        ack.acknowledge();
    //    }

    @KafkaListener(topics = "test_topic", groupId = "testGroup")
    public void batchCustomer(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
        ack.acknowledge();
        records.forEach(record -> {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            kafkaMessage.ifPresent(u -> {
                System.out.println("batch c:" + Thread.currentThread().getName() + ":" + u);
            });
        });
    }
}
